package com.oreilly.learningsparkexamples.scala

import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS
import org.apache.spark.mllib.feature.HashingTF
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 机器学习 - 垃圾邮件分类器
  * Created by cb on 7/27/17.
  */
object EmailSpamML {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(s"Email Spam Machine Learning.")
    val sc = new SparkContext(conf)

    /**
      * 加载２个文件：每１行是为１个邮件
      * spam 为垃圾邮件；
      * ham 为非垃圾邮件。
      */
    val spamFile = args(0)
    val hamFile = args(1)
    val spam = sc.textFile(spamFile)
    val ham = sc.textFile(hamFile)

    //创建１个 HashingTF 实例，用于将电子邮件中的内容转换为 feature vectors，定义100个这样的特征
    val tf = new HashingTF(numFeatures = 100)

    /**
      * 把 mail 分割成 word , 每个 word 转换成1个 feature
      */
    val spamFeatures = spam.map(email => tf.transform(email.split(" ")))
    val hamFeatures = ham.map(email => tf.transform(email.split(" ")))

    //Create LabeledPoint datasets for positive (spam) and negative (ham) examples.
    val positiveExamples = spamFeatures.map(feature => LabeledPoint(1,feature))
    val negativeExamples = hamFeatures.map(feature => LabeledPoint(0,feature))

    val trainingData = positiveExamples ++ negativeExamples
    // Cache data since Logistic Regression is an iterative algorithm.计算成本比较高
    trainingData.cache()

    // 创建 logistic Regression learner 并使用 LBFGS 优化器
    val lrLearner = new LogisticRegressionWithLBFGS()
    //　在训练数据上运行 learning algorithm
    val model = lrLearner.run(trainingData)

    /**
      * 在 positiveExamples 和 negativeExamples 上测试
      * 需要将测试的数据也转换成 feature vectors
      */
    val posTestExamples = tf.transform("Get Viagra real cheap!  Send money right away to ...".split(" "))
    val negTestExamples = tf.transform("Hi Dad, I started studying Spark the other ...".split(" "))

    //使用model去预测上面2个测试邮件是否为垃圾邮件
    println(s"Prediction for positive test example: ${model.predict(posTestExamples)}")
    println(s"Prediction for negative test example: ${model.predict(negTestExamples)}")

    sc.stop()

  }

}
